import MREngine.*;
import x10.util.*;
import x10.io.File;
import x10.io.FileReader;

/** Test Case for MapReduce Engine **/

public class WordCount{
	
	/**
	 * This is a DataBlock used in WordCount application.
	 * The input for WordCount is String.
	 * The DataBlock read the document into the memory.
	 */
	static class WCDataBlock extends MREData{
		
		/** size of the string */
		private var size:Int;
		/** the document content in memory */
		private var content:String;
		
		public def this()
		{}
		// Initial
		
		public def this(var input:String)
		{
			this.size= input.length();
			this.content= input;
		}
		
		public def setContent(input:String)
		{
			this.content= input;
		}
		
		public def getContent()
		{
			return this.content;
		}
		
		public def getSize():Int = size;
	}
	
	/** The raw block data */
	public var data:WCDataBlock;
	/** number of parts to break the raw data into. */
	public var numParts:Int;
	/** number of async */
	public var numAsync:Int;
	
	/** Initial Function reference **/
	private val mres: MapReduceEngineSeq[String,Int];
	private val mrep: MapReduceEnginePar[String,Int];

	
	/** Constructor **/
	public def this(input:String, _np:Int, numAsync:Int)
	{
		data = new WCDataBlock(input);
		numParts = _np;
		mres = new MapReduceEngineSeq[String, Int](data);
	 	mrep = new MapReduceEnginePar[String, Int](data,numAsync);
	}
	
	/** set the split, map, reduce function pointer for MapReduceEngine */
	private def set_smr_func_ptr(engType:String)
	{	
		/** 
		 * Implementation of the split function
		 * data: the raw DataBlock
		 * output: void
		 */
		val _wcSplit=(data:MREData)=>
		{
			var wcdb:WCDataBlock = data as WCDataBlock;
			var content:String = wcdb.getContent();
			splitString (content,engType);
			////Console.OUT.println("DEBUG: split finished");
			return;
		};
		
		/** 
		 * Implementation of the map function 
		 * data: the DataBlock to map
		 * output: void
		 */
		val _wcMap = (data:MREData) => {	
			var wcdb:WCDataBlock = data as WCDataBlock;
			var content:String = wcdb.getContent();
			////Console.OUT.println("DEBUG: content length = " + content.length());
			
			var last_i:Int = 0;
			var i:Int = 0;
			
			while (true)
			{
				while ((i < content.length()-1)			// find the start of a word
						&& !content.charAt(i).isLetter())
					++i;
				if (i == content.length()-1)		// no letter at all
					return;
				
				last_i = i;		// mark the start of a word
				
				while ((i < content.length()-1)			// find end of the word
						&& (content.charAt(i).isLetter()))
					++i;
				
				
				if (i == content.length()-1)		// reach the end
				{
					if(engType.equals("s"))
						mres.EmitIntermediate(content.substring(last_i, i+1), 1);
					else
						mrep.EmitIntermediate(content.substring(last_i, i+1), 1);
	
					return;
				}
				else		// scan a word
				{
					if(engType.equals("s"))
						mres.EmitIntermediate(content.substring(last_i, i), 1);
					else
						mrep.EmitIntermediate(content.substring(last_i, i), 1);
					++i;
				}
			}
		};
		
		/** 
		 * Implementation of the reduce function
		 * key: the key of the list of values to be reduced
		 * valueList: the list of values to be reduced
		 */
		val _wcReduce = (key:String, valueList:ArrayList[MREngine.MREEntry[String, Int]])=>
		{
			var value:Int=0;
			val length = valueList.size();
			////Console.OUT.println("DEBUG: length = : " + length);
			for(var i:Int = 0; i < length; ++i)
			{
				////Console.OUT.println("DEBUG: i = " + i);
				value += valueList.get(i).getValue();
			}
			////Console.OUT.println("DEBUG: one reduce done.");
			if(engType.equals("s"))
				mres.EmitFinal (key, value);
			else
				mrep.EmitFinal (key, value);
			return;
		};
		
		if(engType.equals("s"))
		{
			mres.SetSplitFunc(_wcSplit);
			mres.SetMapFunc(_wcMap);
			mres.SetReduceFunc(_wcReduce);
		}
		else
		{
			mrep.SetSplitFunc(_wcSplit);
			mrep.SetMapFunc(_wcMap);
			mrep.SetReduceFunc(_wcReduce);
		}
		
	}


	/**
	 * Split an input string according to its size and num of parts
	 */ 
	public def splitString(var input:String, engType:String)
	{
		var block_size:Int = data.getSize()/numParts;
		var last_i:Int = 0;
		var i:Int = 0;
		var ch:Char;
		
		//Console.OUT.println("DEBUG: block_size = ["+block_size+"]");
		
		while (true)
		{
			last_i = i;
			i += block_size;
			
			if (i >= input.length()-1)
			{
				val subs = new WCDataBlock(input.substring(last_i, input.length()));
				//Console.OUT.println("DEBUG: subs = ["+ subs.getContent()+"]");
				if(engType.equals("s"))
					mres.EmitSplit(subs);
				else
					mrep.EmitSplit(subs);
				return;
			}
			
			ch = input.charAt(i);
			//Console.OUT.println("DEBUG: ch = ["+ ch +"]");
			while ((i < input.length()-1)
					&& (!ch.isWhitespace()))
			{
				//Console.OUT.println("DEBUG: ch = ["+ ch +"]");
				++i;
				ch = input.charAt(i);
			}
			
			//Console.OUT.println("DEBUG: i = ["+ i +"]");
			//Console.OUT.println("DEBUG: last_i = ["+ last_i +"]");
			//Console.OUT.println("DEBUG: input.length = ["+ input.length() +"]");
			
			if (i == input.length()-1)		// reached the end
			{
				val subs = new WCDataBlock(input.substring(last_i, i+1));
				//Console.OUT.println("DEBUG: subs = ["+ subs.getContent()+"]");
				if(engType.equals("s"))
					mres.EmitSplit (subs);
				else 
					mrep.EmitSplit(subs);
				return;
			}
			else				// found a whitespace
			{
				ch = input.charAt(i);
				while ((i < input.length()-1)		// find the start of the next non-whitespace
						&& (ch.isWhitespace()))
				{
					++i;
					ch = input.charAt(i);
				}
				
				//Console.OUT.println("DEBUG: i = ["+ i +"]");
				if (i == input.length()-1)		
					return;
				
				//Console.OUT.println("DEBUG: before subs = []");
				val subs = new WCDataBlock(input.substring(last_i, i));
				//Console.OUT.println("DEBUG: subs = ["+ subs.getContent()+"]");
				if(engType.equals("s"))
					mres.EmitSplit (subs);
				else
					mrep.EmitSplit(subs);
				//++i;
			}
		}
	}
	

	/** Make a sample text input from a file */
	public static def makeInput(filename:String) 
	{
		var input:String;
		try { 
			val I  = new File(filename);
			
			Console.OUT.println("Size of the file: " + I.size()/1024 + " KB.");
			
			var inputArr:Array[Byte] = new Array[Byte](I.size() as Int);
			
			var i:Int = 0;
			for (byte in I.bytes())
			{
				inputArr(i) = byte;
				++i;
			}
			
			var tempInput:String = new String(inputArr, 0, inputArr.size);
			input = tempInput.toLowerCase();
			
			tempInput = null;
			
			//Console.OUT.println("DEBUG: print out input string: <START>{ " + input + "}<END>");
		} 
		catch ( e2 : x10.io.FileNotFoundException ) { 
			Console.OUT.println("ERROR - File reading failed : " + filename);
			return null;
		}
		return input;
	}

	private def validate():Boolean
	{
		var rSeq:ArrayList[MREngine.MREEntry[String, Int]] = mres.GetResults();
		var rPar:ArrayList[MREngine.MREEntry[String, Int]] = mrep.GetResults();
		val size = rSeq.size();
		
		if (rSeq.size() != rPar.size())
		{
			Console.OUT.println("Seq Size:"+rSeq.size() +" Par Size:" +  rPar.size());
			return false;
		}
		
		var ens:MREngine.MREEntry[String,Int];
		var enp:MREngine.MREEntry[String, Int];
		for(i in 0..(size-1))
		{
			enp = rPar.get(i);
			ens = mres.FindByKey(enp.getKey());
			
			if (ens != null)
			{
				if(enp.getValue() != ens.getValue())
				{
					Console.OUT.println("Inconsist: SVal: " + ens.getValue() +"\tSKey:" + ens.getKey());
					return false;
				}
			}
			else
			{
				Console.OUT.println("Inconsist: Cannot find in serial PVal: " 
						+ enp.getValue() +"\tPKey:" + enp.getKey());
				return false;
			}
		}
		return true;
	}
	
	/**
	 * Print out the result
	 */
	public def print_results ():void
	{
		var results:ArrayList[MREngine.MREEntry[String, Int]] = mres.GetResults();
		var resultp:ArrayList[MREngine.MREEntry[String, Int]] = mrep.GetResults();
		
		Console.OUT.println("Sequential WordCount Results:");
		for (var i:Int = 0; i < results.size()-1; ++i)
			Console.OUT.println( results.get(i).getValue()+ "\t"+results.get(i).getKey());
		
		Console.OUT.println("Parallel WordCount Results:");
		for (var i:Int = 0; i < resultp.size()-1; ++i)
			Console.OUT.println(resultp.get(i).getValue()+ "\t" +resultp.get(i).getKey());

		Console.OUT.println("Total Number of Words: " + results.size());
	}
	
	// for timing
	static val Meg = 1000*1000;
	static val PARTITION_FACTOR = 50;
	
	public static def main(args:Array[String]) {
		
		if (args.size < 4) 
		{
			Console.OUT.println("Usage: x10 WordCount <DocPath:String> <numAsync:Int> <numTrials:Int> <func|perf>");
			return;
		}
		
		if (!args(3).equals("perf") && !args(3).equals("func"))
		{
			Console.OUT.println("Fourth arguement should be 'perf' or 'func'.");
			return;
		}
		
		val inputfile:String = args(0);  // "/Users/sun/Programs/X10/mre/app/test_doc.txt"; //"f:/word.txt"; 
		val input = WordCount.makeInput(inputfile);
		if (input == null)
			return;
		
		val numAsync:Int= Int.parseInt(args(1));	// "24"
		val part:Int = numAsync * PARTITION_FACTOR;	// 2000;	
		val numTrials = Int.parseInt(args(2));	// 2;
		val option:String = args(3);		// "func";
		
		var serialTime:Array[Long] = new Array[Long](numTrials);
		var parallelTime:Array[Long] = new Array[Long](numTrials);
		var speedup:Float = 0;
		
		val wc = new WordCount(input, part, numAsync);
		wc.set_smr_func_ptr("s");
		wc.set_smr_func_ptr("p");

		// functional run 
		if (option.equals("func"))
		{
			Console.OUT.println( "Functionality Test Run Seq, ... " );
			wc.mres.MapReduceSchedulerSerial();
			Console.OUT.println( "Functionality Test Run Par, ... " );
			wc.mrep.MapReduceSchedulerPar();
			Console.OUT.println( "Functionality Test Done. " );

			wc.print_results();
		
			Console.OUT.println( "Validating results... " );
			if(wc.validate())
				Console.OUT.println("Test pass!");
			else
			{
				Console.OUT.println("************************ FAILED! ************************");
				return;
			}
		}
		
		// performance runs
		if (option.equals("perf"))
		{
			var timeSeq:Long;
			var timePar:Long;

			Console.OUT.println( "Performance Runs ... " );
			for ( t in 0..(numTrials-1) ) { 
				Console.OUT.println("Trial " + t + "...");
				
				timeSeq = System.nanoTime();
				wc.mres.MapReduceSchedulerSerial();
				serialTime(t) = (System.nanoTime()-timeSeq)/Meg;			

				timePar = System.nanoTime();
				wc.mrep.MapReduceSchedulerPar();
				parallelTime(t) = (System.nanoTime()-timePar)/Meg;
				
				if (t == 0)
				{
					Console.OUT.println( "Validating results... " );
					if(wc.validate())
						Console.OUT.println("Test pass!");
					else
					{
						Console.OUT.println("************************ FAILED! ************************");
						return;
					}
				}
			}
			
			var avgSerialTime:Long = 0;
			var avgParallelTime:Long = 0;
			
			for (var i:Int = 0; i < serialTime.size; ++i)
			{
				avgSerialTime += serialTime(i);
				avgParallelTime += parallelTime(i);
			}
			
			avgSerialTime /= numTrials;
			avgParallelTime /= numTrials;
			
			speedup = (avgParallelTime > 0) ? (avgSerialTime as Float)/avgParallelTime : Float.MAX_VALUE;
			
			Console.OUT.println("[Done.] Over " + numTrials + " trials, average time" 
					+ " to compute serially is " + avgSerialTime
					+ ",\n    average time to compute parallel is " + avgParallelTime
					+ ",\n    speedup is "+ String.format("%.2f", new Array[Any](1, speedup))+".");
		}

		Console.OUT.println("Word Counting Completed Here.");
	}
}
